package com.cw.realtime.dim
package function

import com.cw.realtime.common.bean.TableProcessDim
import com.cw.realtime.common.constant.Constant.HBASE_NAMESPACE
import com.cw.realtime.common.util.{HbaseUtil, RedisUtil}
import io.circe.JsonObject
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.client.Connection

class HbaseSinkFunction extends RichSinkFunction[(JsonObject, TableProcessDim)] {
  var conn = Option.empty[Connection]

  override def open(parameters: Configuration): Unit =
    conn = HbaseUtil.getConnection

  override def close(): Unit = HbaseUtil.closeConnection(conn)

  override def invoke(value: (JsonObject, TableProcessDim), context: SinkFunction.Context): Unit = {

    for {
      connection <- conn
      _dataType <- value._1("type")
      dataType <- _dataType.as[String]
      _data <- value._1("data")
      data <- _data.asObject
      tblDim = value._2
      _rowKey <- data(tblDim.sinkRowKey)
      rowKey = if (_rowKey.isString) _rowKey.asString.get else _rowKey.toString()
    } {
      if ("delete" == dataType) {
        HbaseUtil.deleteCells(connection, HBASE_NAMESPACE, tblDim.sinkTable, rowKey)
      } else {
        println(
          s""" <<<<<<<<start dml>>>>>>>>>>
             |${value._1}
             |
             |""".stripMargin)
        println(s"family===>${tblDim.sinkFamily}")
        HbaseUtil.putCells(connection, HBASE_NAMESPACE, tblDim.sinkTable, rowKey, tblDim.sinkFamily, _data)
      }

      if ("delete" == dataType || "update" == dataType) {
        val key = RedisUtil.getRedisKey(tblDim.sinkTable, rowKey)
        RedisUtil.getJedis.foreach { j => j.del(key); println("删除缓存====>" + key); j.close() }
      }

    }
  }

}
